package com.skynet.amqp.perfect;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.Instant;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class RetryCache {

    private static final Logger log = LoggerFactory.getLogger(RetryCache.class);

    @Autowired
    private RetryCacheProperties retryCacheProperties;

    private MessageSender messageSender;

    private Map<Long, MessageWrapper> map = new ConcurrentHashMap<>();

    private AtomicLong id = new AtomicLong();

    public long generateId() {
        return id.getAndIncrement();
    }

    public void add(MessageWrapper messageWrapper) {
        map.putIfAbsent(messageWrapper.getId(), messageWrapper);
    }

    public void del(long id) {
        map.remove(id);
    }

    public MessageWrapper get(long id) {
        return map.get(id);
    }

    public void setSender(MessageSender messageSender) {
        this.messageSender = messageSender;
        startRetry();
    }

    private void startRetry() {
        ExecutorService service = Executors.newSingleThreadExecutor();
        RetryCacheProperties.Retry retry = retryCacheProperties.getRetry();
        service.execute(() -> {
            while (retry.retryEnabled) {
                try {
                    TimeUnit.MILLISECONDS.sleep(retry.retryInterval.toMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                long now = Instant.now().toEpochMilli();

                for (Map.Entry<Long, MessageWrapper> entry : map.entrySet()) {
                    MessageWrapper messageWrapper = entry.getValue();
                    if (messageWrapper != null) {
                        if (messageWrapper.getTime() + retry.retryExpire.toMillis() < now) {
                            log.warn("The message {} is expired,ready to remove from retry cache", messageWrapper);
                            del(entry.getKey());
                        } else if (messageWrapper.getTime() + retry.retryDelay.toMillis() < now) {
                            log.info("Send Message retrying...");
                            messageSender.send(messageWrapper);
                        }
                    }
                }
            }

        });
    }

}
